package com.czl.dxpro.dx_11_taskEleven;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class DxElevenMain {

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		// TODO Auto-generated method stub
		Configuration conf=new Configuration();
		conf.set("cyc", "3");
		Job job= Job.getInstance();
		job.setJarByClass(DxElevenMain.class);
		job.setMapperClass(ElevenMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		
		job.setReducerClass(ElevenReduce.class);
		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(Text.class);
		
		
		FileSystem fs=FileSystem.get(new Configuration());
		//"g://data/dx"
		FileStatus[] status=fs.listStatus(new Path("d://data/dx"));
		String pathname="";
		int a=0;
		for(FileStatus file :status) {
			String pname=file.getPath().toString();
			if(pname.matches(".*\\/(\\d{8})$")&&!pname.endsWith(args[1])) {
				String[] str=pname.split("\\/");
				if(Integer.parseInt(str[str.length-1])>a&&Integer.parseInt(str[str.length-1])<Integer.parseInt(args[1])){
					a=Integer.parseInt(str[str.length-1]);
					pathname=pname;
				}
			}
		}
		if(!"".equals(pathname)) {
			FileInputFormat.addInputPath(job, new Path(pathname+"/livedataout"));
		}
		//"g://data/dx/Dxcollect/"
		FileInputFormat.addInputPath(job, new Path("d://data/dx/DxTen"));
		//"g://data/dx/Dxlive/livedataout"
		FileOutputFormat.setOutputPath(job, new Path("d://data/dx/DxEleven"));
		System.exit(job.waitForCompletion(true)?0:1);
	}

}

